package com.atguigu.education.app.dwd.log;

import com.alibaba.fastjson.JSONAware;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.education.util.KafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * @Author luyang
 * @create 2022/9/4 14:01
 */
public class DwdUserBehavior {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //设置并行度
        env.setParallelism(1);

//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2 设置检查点和状态后端
        /*     //先开启  多长时间去设置一个状态后端 检查点模式一般使用精准一次
        env.enableCheckpointing(5*60*1000L, CheckpointingMode.EXACTLY_ONCE);
        //超时时间
        env.getCheckpointConfig().setCheckpointTimeout(3*60*1000L);
        //最多可以出现几个状态后端
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        //状态保存位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall/ck");
        //上面保存位置如果是在hdfs上的话 那么需要用户名
        System.setProperty("HADOOP_USER_NAME","atguigu");
        //开一下状态后端
        env.setStateBackend(new HashMapStateBackend());
        */
        //TODO 3 读取kafka数据
        String topicName = "dwd_traffic_page_log";
        String groupID = "dwd_user_behavior";

        DataStreamSource<String> topicLogStream = env.addSource(KafkaUtil.getKafkaConsumer(topicName, groupID));
        //topicLogStream.print("log>>>>>>");

        // TODO 4 过滤转换
        SingleOutputStreamOperator<JSONObject> flatMapStream = topicLogStream.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSONObject.parseObject(value);
                JSONObject page = jsonObject.getJSONObject("page");
                if ("home".equals(page.getString("page_id")) || "course_detail".equals(page.getString("page_id"))) {
                    out.collect(jsonObject);
                }
            }
        });

        // todo 7、将数据传回到Kafka中
        String targetTopic = "dwd_user_behavior";
        flatMapStream.map(JSONAware::toJSONString)
                .addSink(KafkaUtil.getKafkaProducer(targetTopic));

        //TODO 执行任务
        env.execute();
    }
}
